package com.etc

import org.apache.spark.{SparkConf, SparkContext}

/**
  * @Auther: Wangcc
  * @Date: 2018/8/17 19:14
  * @Description:
  */
object ActionOperationScala {
  def main(args: Array[String]) {
    // reduce()
    // collect()
    // count()
    // take()
    countByKey()
  }

  def reduce() {
    val conf = new SparkConf()
      .setAppName("reduce")
      .setMaster("local")
    val sc = new SparkContext(conf)

    val numberArray = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val numbers = sc.parallelize(numberArray, 1)
    val sum = numbers.reduce(_ + _)

    println(sum)
  }

  def collect() {
    val conf = new SparkConf()
      .setAppName("collect")
      .setMaster("local")
    val sc = new SparkContext(conf)

    val numberArray = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val numbers = sc.parallelize(numberArray, 1)
    val doubleNumbers = numbers.map { num => num * 2 }

    val doubleNumberArray = doubleNumbers.collect()

    for(num <- doubleNumberArray) {
      println(num)
    }
  }

  def count() {
    val conf = new SparkConf()
      .setAppName("count")
      .setMaster("local")
    val sc = new SparkContext(conf)

    val numberArray = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val numbers = sc.parallelize(numberArray, 1)
    val count = numbers.count()

    println(count)
  }

  def take() {
    val conf = new SparkConf()
      .setAppName("take")
      .setMaster("local")
    val sc = new SparkContext(conf)

    val numberArray = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val numbers = sc.parallelize(numberArray, 1)

    val top3Numbers = numbers.take(3)

    for(num <- top3Numbers) {
      println(num)
    }
  }

  def saveAsTextFile() {

  }

  def countByKey() {
    val conf = new SparkConf()
      .setAppName("countByKey")
      .setMaster("local")
    val sc = new SparkContext(conf)

    val studentList = Array(Tuple2("class1", "leo"), Tuple2("class2", "jack"),
      Tuple2("class1", "tom"), Tuple2("class2", "jen"), Tuple2("class2", "marry"))
    val students = sc.parallelize(studentList, 1)
    val studentCounts = students.countByKey()

    println(studentCounts)
  }

}
